package com.zsk.mqtt.app;

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttTopic;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

/**
 * 消息保留机制https://www.cnblogs.com/smartlife/articles/10166718.html
 * 
 * @author zhangsike
 * @date 2020-05-17 11:51:46
 */
// emqx 用户名appuser admin
// emqx 用户名deviceuser admin
public class Mqtt {

    // 61613
    // public static final String SERVER_URL = "tcp://127.0.0.1:11883";
    public static final String SERVER_URL = "tcp://117.36.74.60:1883";
    // 信息回调
    protected StateListener stateListener;
    // 客户端id
    protected String clientId;
    // MQTT客戶端
    protected MqttClient mqttClient;
    // MQTT的连接设置
    protected MqttConnectOptions mqttConnectOptions;

    protected String username = "admin";
    // protected String password = "password";
    protected String password = "public";

    public Mqtt(String clientId, StateListener listener) {
        this.clientId = clientId;
        this.stateListener = listener;
        this.initConnectOption();
        this.initMqttClient();

    }

    protected void initMqttClient() {
        try {
            mqttClient = new MqttClient(SERVER_URL, clientId, new MemoryPersistence());
            mqttClient.setCallback(new MqttCallbackExtended() {

                @Override
                public void messageArrived(String topic, MqttMessage message) throws Exception {
                    // subscribe后得到的消息会执行到这里面
                    if (stateListener != null) {
                        stateListener.messageArrived(topic, message);
                    }
                    System.out
                        .println("收到消息" + topic + "," + message.getQos() + "," + new String(message.getPayload()));
                }

                @Override
                public void deliveryComplete(IMqttDeliveryToken token) {
                    // publish后会执行到这里
                    if (stateListener != null) {
                        stateListener.deliveryComplete(token);
                    }
                    // System.out.println("发送完成");
                }

                @Override
                public void connectionLost(Throwable cause) {
                    System.out.println("连接断开");
                    // 一般在这里面进行重连
                    if (stateListener != null) {
                        stateListener.connectionLost(cause);
                    }
                }

                @Override
                public void connectComplete(boolean reconnect, String serverURI) {
                    // 连接成功后调用
                    // 在此出实现订阅
                    if (stateListener != null) {
                        stateListener.connectComplete(reconnect, serverURI);
                    }

                }

            });
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    protected void initConnectOption() {
        mqttConnectOptions = new MqttConnectOptions();
        // 自动重连
        mqttConnectOptions.setAutomaticReconnect(true);
        // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，这里设置为true表示每次连接到服务器都以新的身份连接
        mqttConnectOptions.setCleanSession(false);
        // mqtt服务器用户名
        mqttConnectOptions.setUserName(username);
        // mqtt服务器密码
        mqttConnectOptions.setPassword(password.toCharArray());
        // 设置超时时间单位为秒
        mqttConnectOptions.setConnectionTimeout(5);
        // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
        mqttConnectOptions.setKeepAliveInterval(20);
    }

    /**
     * 连接服务器
     * 
     * @param topic
     * @throws MqttException
     */
    public void connect() throws MqttException {
        mqttClient.connect(mqttConnectOptions);
    }

    /**
     * 判断是否连接
     * 
     * @return
     * @throws MqttException
     */
    public boolean isConnected() throws MqttException {
        if (mqttClient != null) {
            return mqttClient.isConnected();
        }
        return false;
    }

    /**
     * 断开连接
     * 
     * @throws MqttException
     */
    public void disconnect() throws MqttException {
        if (mqttClient != null) {
            mqttClient.disconnect();
            mqttClient = null;
        }
    }

    /**
     * 重新连接
     * 
     * @throws MqttException
     */
    public void reconnect() throws MqttException {
        if (mqttClient != null) {
            mqttClient.reconnect();
        }
    }

    /**
     * 发送主题消息
     * 
     * @param topic
     *            主题
     * @param msg
     *            消息
     * @param qos
     *            消息质量 0，1，2
     * @param retained
     *            如果PUBLISH消息的RETAIN标记位被设置为1，则称该消息为“保留消息”；Broker会 存储每个Topic的最后一条保留消息
     *            及其Qos，当订阅该Topic的客户端上线后，Broker需要将该消息投递给它。
     * @throws @throws
     *             MqttException
     */
    public void publish(String topic, String msg, int qos, boolean retained) throws MqttException {
        mqttClient.publish(topic, msg.getBytes(), qos, retained);
    }

    public void publish(MqttTopic topic, MqttMessage message) throws MqttException {
        topic.publish(message).waitForCompletion();
    }

    /**
     * 订阅主题
     * 
     * @param topic
     * @param qos
     * @throws MqttException
     */
    public boolean subscribe(String topic, int qos) throws MqttException {
        if (mqttClient == null) {
            return false;
        }
        mqttClient.subscribe(topic, qos);
        return true;
    }

    /**
     * 取消订阅主题
     * 
     * @param topic
     * @param qos
     * @throws MqttException
     */
    public boolean unsubscribe(String topic) throws MqttException {
        if (mqttClient == null) {
            return false;
        }
        mqttClient.unsubscribe(topic);
        return true;
    }

    /**
     * 回调接口
     * 
     * @author zhangsike
     * @date 2020-05-16 15:44:43
     */
    public interface StateListener {

        /**
         * 消息到达时
         * 
         * @param topic
         * @param message
         */
        public void messageArrived(String topic, MqttMessage message);

        /**
         * 
         * @param token
         */
        public void deliveryComplete(IMqttDeliveryToken token);

        /**
         * 连接断开时
         * 
         * @param cause
         */
        public void connectionLost(Throwable cause);

        /**
         * 连接完成时
         * 
         * @param reconnect
         * @param serverURI
         */
        public void connectComplete(boolean reconnect, String serverURI);

    }
}